package com.zhang.flink.example.source;

import com.zhang.flink.bean.ClickEvent;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.Calendar;
import java.util.Random;

/**
 * @title:
 * @author: zhang
 * @date: 2022/2/11 20:33
 */
public class ClickSource implements SourceFunction<ClickEvent> {
    private boolean running = true;
    private Random random = new Random();
    private String[] userArr = {"zhangsan","bob","Alice","John","Mary"};
    private String[] urlArr = {"./home","./fav","./cart","./order","./prod?id=2"};

    @Override
    public void run(SourceContext<ClickEvent> ctx) throws Exception {
        while (running){
            ctx.collect(new ClickEvent(
                    userArr[random.nextInt(userArr.length)],
                    urlArr[random.nextInt(urlArr.length)],
                    Calendar.getInstance().getTimeInMillis()
            ));
            Thread.sleep(500L);
        }
    }

    @Override
    public void cancel() {
        running = false;
    }
}
